package com.nowcoder.community.event.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.nowcoder.community.entity.SystemMessage;
import com.nowcoder.community.event.CommentEvent;
import com.nowcoder.community.event.FollowEvent;
import com.nowcoder.community.event.LikeEvent;
import com.nowcoder.community.event.producer.DefaultEventProducer;
import com.nowcoder.community.service.MessageService;
import com.nowcoder.community.util.CommunityUtils;
import com.nowcoder.community.util.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;

/**
 * 处理SystemNoticeProducer中产生的消息
 * @see DefaultEventProducer
 */
@Slf4j
@Component
public class SystemNoticeConsumer {


    @Autowired
    private MessageService messageService;

    private String receiveMessage(ConsumerRecord<String, Object> record) {
        String msg = (String) record.value();
        log.debug("收到消息 FROM <{}>: {}",record.topic(), msg);
        return msg;
    }

    @KafkaListener(topics = { Constants.USER_COMMENT_TOPIC })
    public void handleComment(ConsumerRecord<String, Object> record) {
        try {
            CommentEvent event = CommunityUtils.parseJson(receiveMessage(record), CommentEvent.class);
            String content = CommunityUtils.toJson(new HashMap<String, Object>() {{
                put("entityId", event.getEntityId());
                put("entityType", event.getEntityType());
                put("postId", event.getPostId());
                put("userId", event.getCommentUserId());
            }});
            SystemMessage message = new SystemMessage(event.getUserIdOfEntity(), Constants.SYSTEM_NOTICE_COMMENT);
            message.setCreateTime(new Date(event.getCreateTime()));
            message.setContent(content);
            messageService.sendCommentSystemNotice(message);
        } catch (JsonProcessingException e) {
            log.error("无法处理CommentEvent消息, 消息处理失败, message: {}, Cause: {}", record.value(), e.getMessage());
        }
    }

    @KafkaListener(topics = { Constants.USER_LIKE_TOPIC })
    public void handleLike(ConsumerRecord<String, Object> record) {
        try {
            LikeEvent event = CommunityUtils.parseJson(receiveMessage(record), LikeEvent.class);
            String content = CommunityUtils.toJson(new HashMap<String, Object>() {{
                put("entityId", event.getEntityId());
                put("entityType", event.getEntityType());
                put("userId", event.getLikeUserId());
            }});
            SystemMessage message = new SystemMessage(event.getUserIdOfEntity(), Constants.SYSTEM_NOTICE_LIKE);
            message.setCreateTime(new Date(event.getCreateTime()));
            message.setContent(content);
            messageService.sendCommentSystemNotice(message);
        } catch (JsonProcessingException e) {
            log.error("无法处理LikeEvent消息, 消息处理失败, message: {}, Cause: {}", record.value(), e.getMessage());
        }
    }

    @KafkaListener(topics = { Constants.USER_FOLLOW_TOPIC })
    public void handleFollow(ConsumerRecord<String, Object> record) {
        try {
            FollowEvent event = CommunityUtils.parseJson(receiveMessage(record), FollowEvent.class);
            String content = CommunityUtils.toJson(new HashMap<String, Object>() {{
                put("userId", event.getFollowerId());
            }});
            SystemMessage message = new SystemMessage(event.getUserId(), Constants.SYSTEM_NOTICE_FOLLOW);
            message.setCreateTime(new Date(event.getCreateTime()));
            message.setContent(content);
            messageService.sendCommentSystemNotice(message);
        } catch (JsonProcessingException e) {
            log.error("无法处理FollowEvent消息, 消息处理失败, message: {}, Cause: {}", record.value(), e.getMessage());
        }
    }
}
